package com.ztorn.cdc.pulsar;

import com.ztorn.cdc.AbstractSinkBuilder;
import com.ztorn.cdc.CDCBuilder;
import com.ztorn.cdc.SinkBuilder;
import com.ztorn.common.model.FlinkCDCConfig;
import com.ztorn.common.model.Table;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.LogicalType;

import java.io.Serializable;
import java.util.List;

public class PulsarSinkBuilder extends AbstractSinkBuilder implements SinkBuilder, Serializable {

    public static final String KEY_WORD = "datastream-pulsar";

    public PulsarSinkBuilder(){}

    public PulsarSinkBuilder(FlinkCDCConfig config){
        super(config);
    }


    @Override
    public void addSink(StreamExecutionEnvironment env, DataStream<RowData> rowDataDataStream, Table table, List<String> columnNameList, List<LogicalType> columnTypeList) {

    }

    @Override
    public String getHandle() {
        return KEY_WORD;
    }

    @Override
    public SinkBuilder create(FlinkCDCConfig config) {
        return new PulsarSinkBuilder(config);
    }

    @Override
    public DataStreamSource build(CDCBuilder cdcBuilder,StreamExecutionEnvironment env,DataStreamSource<String> dataStreamSource) {
        PulsarSinkFunction sinkFunction = new PulsarSinkFunction<>(
                config.getSink().get("topics"),
                config.getSink().get("serviceUrl"),
                new SimpleStringSchema());
        dataStreamSource.addSink(sinkFunction);
        return dataStreamSource;
    }
}
